package com.junlinpro.service;


import java.io.FileOutputStream;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import com.alibaba.fastjson.JSONObject;
import com.junlinpro.config.ApplicationContextProvider;
import com.junlinpro.entity.MessageEntity;
import com.junlinpro.entity.RegisterEntity;
import com.junlinpro.utils.ByteUtil;
import com.junlinpro.utils.Constant;
import com.junlinpro.utils.CrcUtils;
import com.junlinpro.utils.DateUtil;
import com.junlinpro.utils.HBaseUtil;
import com.junlinpro.utils.KafkaProducerUtil;
import com.junlinpro.utils.MessageUtil;
import com.junlinpro.utils.PcmInByteData;
import com.junlinpro.utils.SocketChannelMap;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class RealTimeAudioService {

    private InetSocketAddress inetSocketAddress = new InetSocketAddress("10.1.52.1", 9080);
    private DatagramSocket datagramSocket = null;
    private HashMap<String,Integer> cardIdMap = new HashMap<String,Integer>();
    private PcmInByteData pcmInByteData;
    
    public RealTimeAudioService() {
    	this.pcmInByteData = ApplicationContextProvider.getBean(PcmInByteData.class);
    	try {
			datagramSocket = new DatagramSocket();
		} catch (SocketException e) {			
			e.printStackTrace();
		}
    }	    
	
   /**
	 *    存入kafka 并发给前端
	* @param soundData
	*/
	@Async
	public void sendSoundToUDP(MessageEntity msgEntity,byte[] soundData) {
		int frameId = msgEntity.getFrameId();
		
		int count = (frameId /Constant.REALTIME_FRAME_SUM)+1;
		cardIdMap.put(msgEntity.getCardId()+"_"+msgEntity.getChannelId(), count);
		
		StringBuilder rowKey = new StringBuilder();
		rowKey.append(msgEntity.getCardId());		
        rowKey.append("_");
        rowKey.append(msgEntity.getChannelId());
        rowKey.append("_");
        rowKey.append(msgEntity.getCardType());
        rowKey.append("_");
        rowKey.append(msgEntity.getOccurTime());
        rowKey.append("_");
        
        //1 实时监听每帧数据存入kafka 以便算法实时分析
		KafkaProducerUtil.saveRealTimeSound(rowKey.toString()+count, soundData);
		
		//2 实时监听的时长达到20分钟则存入hbase，以免大量占用内存
		if( frameId>0  && frameId %50 == 0) {
			log.info("KafkaProducerUtil.saveRealTimeSound:"+(rowKey.toString()+count));
			//250帧数据时长20秒      250*10*6 =15000= 20分钟   
			if(frameId %(Constant.REALTIME_FRAME_SUM) == 0) {
				
				savePcmDataToHbase(msgEntity,rowKey.toString()+(count-1),(count-1));
			}
		}	
		
		//3 实时监听每帧数据发送web端，以便页面实时播放
		//3.1前5个字节存储cardId和channelId
		byte[] cardIdBytes = ByteUtil.intToByteArray(msgEntity.getCardId());
		byte[] finalData = new byte[soundData.length+5];
		finalData[0] = cardIdBytes[0];
		finalData[1] = cardIdBytes[1];
		finalData[2] = cardIdBytes[2];
		finalData[3] = cardIdBytes[3];
		finalData[4] = msgEntity.getChannelId();
		//3.2 其余字节存储音频数据
		int i = 5;
		for (byte cc : soundData) {
		finalData[i] = cc;
		i++;
		}
		
		//3.3 实时监听每帧数据发送web端
		DatagramPacket datagramPacket = null;
		try {        	
			datagramPacket = new DatagramPacket(finalData, finalData.length, inetSocketAddress);
			datagramSocket.send(datagramPacket);
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
		
		}
	}
	
    public void saveRealTimeData(MessageEntity msgEntity) {
	
    	StringBuilder rowKey = new StringBuilder();
    	rowKey.append(msgEntity.getCardId());	
        rowKey.append("_");
        rowKey.append(msgEntity.getChannelId());
        rowKey.append("_");
        rowKey.append(msgEntity.getCardType());
        rowKey.append("_");
        rowKey.append(msgEntity.getOccurTime());
        rowKey.append("_");         
        String pcmKey = msgEntity.getCardId() + "_" + msgEntity.getChannelId();
        //拼接出这是第几段音频
        rowKey.append(cardIdMap.get(pcmKey)); 
        
        savePcmDataToHbase(msgEntity,rowKey.toString(),cardIdMap.get(pcmKey));
        
        //cardIdMap是记录采集卡实时监听的分段存储总数，实时上报结束时删除
        cardIdMap.remove(pcmKey); 
    } 
    
    /**
               *    用于实时监听定时保存数据  避免内存数据过大，内存溢出
     * @param cardId
     * @param channelId
     * @param cardType
     * @param utcTime
     */
    public void savePcmDataToHbase(MessageEntity msgEntity,String rowKey,int count) {
    	Map<String, byte[]> credByteMap = new HashMap<>();
 
    	String pcmKey = msgEntity.getCardId() + "_" + msgEntity.getChannelId()+count; 

        List<byte[]> byteList = pcmInByteData.get(pcmKey);
        if (byteList != null && !byteList.isEmpty()) {
            // 缓存中的数据
            log.info(" savePcmDataToHbasedata  byteList.size()="+byteList.size());
            // 循环取出缓存中的数据
            for (byte[] pcmData : byteList) {
            	 
            	//在tcp接收时已解析，此处故不需要再opusSdk解码
                 //存在则追加数据
                if (credByteMap.get(pcmKey) != null) {
                    int currentSize = credByteMap.get(pcmKey).length;
                    byte[] currentData = credByteMap.get(pcmKey);
                    int finalSize = currentSize + pcmData.length;
                    byte[] finalData = new byte[finalSize];
                    int g = 0;
                    for (byte cc : currentData) {
                        finalData[g] = cc;
                        g++;
                    }
                    for (byte cc : pcmData) {
                        finalData[g] = cc;
                        g++;
                    }
                    credByteMap.put(pcmKey, finalData);
                } else {
                    credByteMap.put(pcmKey, pcmData);
                }
            }
            
            byte[] bytes = credByteMap.get(pcmKey);
            log.info("Final data size:"+bytes.length);
            
            try {
                toWavFile(bytes);
            }catch (IOException e) {
                e.printStackTrace();
            }
            
            try {            	
                log.info("实时监听，将音频数据存入hbase,rowKey="+rowKey);
            	HBaseUtil.saveSoudataTimeorder(msgEntity,rowKey, bytes);                
            }catch (Exception e) {
            	log.error(e.getMessage());
            	e.printStackTrace();
            }            
            
        }else {
        	log.info("pcmInByteData.get(pcmKey) is null,pcmKey="+pcmKey);
        }        
        // 将原缓存中的数据清除掉
        pcmInByteData.removePcmData(pcmKey);
    }

    @Async
    public void register(ChannelHandlerContext ctx, RegisterEntity registerEntity, int registerNum) {
    	
        MessageEntity replyMsg = new MessageEntity();
        replyMsg.setCardId(registerEntity.getCardId());
        replyMsg.setChannelId(registerEntity.getChannelId());        
        
        RestTemplate restTemplate = new RestTemplate();      
        
        //cmd默认为注册查询
        byte cmd= 0x01;
        byte[] byteData = null;
        if(registerNum==1){
        	// 调用web端注册查询接口
        	String url = Constant.REGISTER_QUERY_URL+registerEntity.getCardId();
        	ResponseEntity<JSONObject> response = restTemplate.postForEntity(url, null, JSONObject.class);
        	JSONObject jsonObject = response.getBody(); 
        	
        	//{code:200,msg:”采集卡未注册”,data:0}{code:200,msg:”采集卡已注册”,data:1}
        	if(jsonObject.getIntValue("code")==200) {
        		if(jsonObject.getIntValue("data")==1) {
        			byteData = new byte[]{0x02};  //已注册
        		}else if(jsonObject.getIntValue("data")==0) {
        			byteData = new byte[]{0x01}; //未注册
        		}else {
        			log.info("注册查询接口返回未知值");
        		}        		
        	}else {
        		log.info("注册查询接口调用失败");            	
            }
        	
        }else if(registerNum==2){
        	//注册请求
        	cmd = 0x0B;
        	
        	// 调用web端注册请求接口
        	String url = Constant.REGISTER_URL+registerEntity.getCardId();
        	ResponseEntity<JSONObject> response = restTemplate.postForEntity(url, null, JSONObject.class);
        	JSONObject jsonObject = response.getBody();
        	
        	//{code:200,msg:”操作成功”} {code:201,msg:”此采集卡已注册”}
        	if(jsonObject.getIntValue("code")==200) {        		
        		byteData = new byte[]{0x03}; //注册成功         		
        	}else if(jsonObject.getIntValue("code")==201){        		
        		byteData = new byte[]{0x03}; //已注册        		          	
            }else {
            	byteData = new byte[]{0x04}; //注册失败或已注册
            	log.info("注册接口调用失败");  
            }        	
        	cmd = 0x0B;        	
        } 
        replyMsg.setData(byteData);
        replyMsg.setCmd(cmd);
        //固定23+数据1个字节
        replyMsg.setAllLen(24);
        replyMsg.setVersionNum(registerEntity.getVersionNum());

        // 帧序号
        replyMsg.setFrameId(0);
        // 类型
        byte cardType  = 0x01;
        if (registerEntity.getCardType() == 1) {
            cardType = 0x00;
        }
        replyMsg.setCardType(cardType);

        // 转换成时间戳
        int utcTimeStamp = DateUtil.transForMilliSecond(new Date());
        replyMsg.setOccurTime(utcTimeStamp);

        // 对回复信息封装
        byte[] byteMsg = MessageUtil.rtnMessageTobyte(replyMsg);

        log.info("注册：" + ByteUtil.bytesToHexString(byteMsg));

        log.info("channel：" + ctx.toString());

        try {
            // 回复信息
            ctx.writeAndFlush(Unpooled.copiedBuffer(byteMsg));

        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    @Async
    public void startCmd(Channel channel, String cardId, String channelId, int cmdType) {
        MessageEntity msg = new MessageEntity();
        byte[] cardidByte = CrcUtils.intToBytesTwo(Integer.parseInt(cardId));
        msg.setCardId(CrcUtils.bytesToInt(cardidByte));
        msg.setChannelId(Byte.parseByte(channelId));

        // 任务id
        String taskKey = "";
        byte cmd = 0x00;
        if (cmdType == 1) {
            cmd = 0x03;
            taskKey = SocketChannelMap.startTaskKey + cardId;
        } else if (cmdType == 2) {
            cmd = 0x04;
            taskKey = SocketChannelMap.stopTaskKey + cardId;
        }
        msg.setCmd(cmd);
        msg.setAllLen(23);
        msg.setVersionNum(Constant.HARDWARE_VERSION);

        int utcTimeStamp = DateUtil.transForMilliSecond(new Date());
        msg.setOccurTime(utcTimeStamp);

        byte[] byteMsg = MessageUtil.rtnMessageTobyte(msg);

        try {
            // 预防嵌入式收不到，持续下发1分钟的命令
            for (int i = 0; i < 30; i++) {
                if (SocketChannelMap.taskMapGet(taskKey)) {
                	//log.info("收到回复停止下发命令");
                    // 如果为 true 说明收到嵌入式的回复，删除内存中数据，并停止循环
                    SocketChannelMap.taskMapRemove(taskKey);
                    //记录采集卡在监听
                    if (cmdType == 1){
                    	SocketChannelMap.monitorTaskMapPut(cardId,true);
                    }else {
                    	SocketChannelMap.monitorTaskMapRemove(cardId);
                    }
                    break;
                } else {
                	log.info(cmd+":"+cardId+"实时监听指令重复下发---"+i);
                    // 下发指令
                    channel.writeAndFlush(Unpooled.copiedBuffer(byteMsg));
                    Thread.sleep(2000);
                }
            }
        } catch (Exception e) {
        	log.info("--- 实时监听下发命令线程出错 -- -" +cmd);
            e.printStackTrace();
        }
    } 
    
    private void toWavFile(byte[] bytes) throws IOException {
        String fileDic = "/mnt/audiofile/";
        String fileName = "realtime_"+UUID.randomUUID().toString()+".wav";
        String targetFile = fileDic+fileName;

        FileOutputStream fos = new FileOutputStream(targetFile);
        long size = bytes.length;
        int totalDataLen = new Long(size).intValue();
        int totalLen = new Long(size).intValue()+44;
        ByteBuffer buffer = ByteBuffer.allocate(new Long(size).intValue()+44);
        byte[] header = getHeader(totalDataLen, totalLen, 16000, 1, 16000 * 2);
        buffer.put(header);

        buffer.put(bytes);

        buffer.flip();
        byte[] data = new byte[totalLen];
        buffer.get(data);
        fos.write(data);
    }

    private byte[] getHeader(long totalAudioLen,
                             long totalDataLen, long longSampleRate, int channels, long byteRate) {
        byte[] header = new byte[44];
        header[0] = 'R'; // RIFF/WAVE header
        header[1] = 'I';
        header[2] = 'F';
        header[3] = 'F';
        header[4] = (byte) (totalDataLen & 0xff);
        header[5] = (byte) ((totalDataLen >> 8) & 0xff);
        header[6] = (byte) ((totalDataLen >> 16) & 0xff);
        header[7] = (byte) ((totalDataLen >> 24) & 0xff);
        header[8] = 'W';
        header[9] = 'A';
        header[10] = 'V';
        header[11] = 'E';
        header[12] = 'f'; // 'fmt ' chunk
        header[13] = 'm';
        header[14] = 't';
        header[15] = ' ';
        header[16] = 16; // 4 bytes: size of 'fmt ' chunk
        header[17] = 0;
        header[18] = 0;
        header[19] = 0;
        header[20] = 1; // format = 1
        header[21] = 0;
        header[22] = (byte) channels;
        header[23] = 0;
        header[24] = (byte) (longSampleRate & 0xff);
        header[25] = (byte) ((longSampleRate >> 8) & 0xff);
        header[26] = (byte) ((longSampleRate >> 16) & 0xff);
        header[27] = (byte) ((longSampleRate >> 24) & 0xff);
        header[28] = (byte) (byteRate & 0xff);
        header[29] = (byte) ((byteRate >> 8) & 0xff);
        header[30] = (byte) ((byteRate >> 16) & 0xff);
        header[31] = (byte) ((byteRate >> 24) & 0xff);
        header[32] = (byte) (1 * 16 / 8); // block align
        header[33] = 0;
        header[34] = 16; // bits per sample
        header[35] = 0;
        header[36] = 'd';
        header[37] = 'a';
        header[38] = 't';
        header[39] = 'a';
        header[40] = (byte) (totalAudioLen & 0xff);
        header[41] = (byte) ((totalAudioLen >> 8) & 0xff);
        header[42] = (byte) ((totalAudioLen >> 16) & 0xff);
        header[43] = (byte) ((totalAudioLen >> 24) & 0xff);
        return header;
    }
    

}
